package com.qezhhnjy.hbase.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author zhaoyangfu - 2021/2/4 16:49
 */
@Component
@Slf4j
public class HBaseService {

    private Connection connection;

    public HBaseService(Configuration configuration) {
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            log.error("获取HBase连接失败");
        }
    }

    @PreDestroy
    private void close() throws IOException {
        if (connection != null) connection.close();
    }

    public Admin admin() throws IOException {
        return connection.getAdmin();
    }

    public Table table(String tableName) throws IOException {
        return connection.getTable(TableName.valueOf(tableName));
    }

    public void listTable() {
        try (Admin admin = admin()) {
            admin.listTableDescriptors().forEach(table -> log.info(table.toString()));
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    public void createTable(String tableName, List<String> familyNames) {
        try (Admin admin = admin()) {

            if (admin.tableExists(TableName.valueOf(tableName))) {
                log.warn("table=>{} exists", tableName);
                return;
            }
            TableDescriptor tableDescriptor = TableDescriptorBuilder
                    .newBuilder(TableName.valueOf(tableName))
                    .setColumnFamilies(familyNames.stream().map(ColumnFamilyDescriptorBuilder::of).collect(Collectors.toList()))
                    .build();

            admin.createTable(tableDescriptor);
            log.info("table=>{} create success", tableName);
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 要删除表或改变其设置，首先需要使用 disable 命令关闭表。
     * 禁用表之后，仍然可以通过 list 和 exists 命令查看到,无法扫描到它存在.
     * 使用 enable 命令，可以重新启用它。
     */
    public void disable(String tableName) {
        try (Admin admin = admin()) {
            TableName table = TableName.valueOf(tableName);
            if (admin.isTableEnabled(table)) {
                admin.disableTable(table);
            }
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    public void enable(String tableName) {
        try (Admin admin = admin()) {
            TableName table = TableName.valueOf(tableName);
            if (admin.isTableDisabled(table)) {
                admin.enableTable(table);
            }
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    public void addFamily(String tableName, String family) {
        try (Admin admin = admin()) {
            admin.addColumnFamily(TableName.valueOf(tableName), ColumnFamilyDescriptorBuilder.of(family));
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    public void deleteFamily(String tableName, String family) {
        try (Admin admin = admin()) {
            admin.deleteColumnFamily(TableName.valueOf(tableName), Bytes.toBytes(family));
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    public boolean exists(String tableName) {
        try (Admin admin = admin()) {
            return admin.tableExists(TableName.valueOf(tableName));
        } catch (IOException e) {
            log.error(e.getMessage(), e);
            return false;
        }
    }

    public void deleteTable(String tableName) {
        try (Admin admin = admin()) {
            TableName table = TableName.valueOf(tableName);
            admin.disableTable(table);
            admin.deleteTable(table);
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    public void truncateTable(String tableName, boolean preserveSplits) {
        try (Admin admin = admin()) {
            TableName table = TableName.valueOf(tableName);
            admin.disableTable(table);
            admin.truncateTable(table, preserveSplits);
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }


    /**
     * 为表添加或者更新数据
     */
    public void put(String tableName, String rowKey, String familyName, String[] columns, String[] values) {
        try (Table table = table(tableName)) {
            //设置row key
            Put put = new Put(Bytes.toBytes(rowKey));
            if (columns != null && values != null && columns.length == values.length) {
                for (int i = 0; i < columns.length; i++) {
                    if (columns[i] != null && values[i] != null) {
                        put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columns[i]), Bytes.toBytes(values[i]));
                    } else {
                        log.warn("列名和列数据都不能为空,column:{},value:{}", columns[i], values[i]);
                    }
                }
            }
            table.put(put);
            log.debug("putData add or update data Success,rowKey:" + rowKey);
        } catch (Exception e) {
            log.error("为表添加 or 更新数据失败,tableName:{},rowKey:{},familyName:{}", tableName, rowKey, familyName, e);
        }
    }
}
